/**
 * Copyright (c) 2015 云智盛世
 * Created with Counter.
 */
package com.ecmaster.search.count;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.net.URLDecoder;
import java.util.Arrays;
import java.util.Iterator;

/**
 * Class description
 *
 * @author GanJc on 2015/9/14
 */
public class Counter extends Configured implements Tool {

    public static void main(String[] args) throws Exception {
//        test();
        int run = ToolRunner.run(new Counter(), args);
        System.exit(run);
    }

    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        Job job = new Job(conf, "Load Redis");
        job.setJarByClass(Counter.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(RedisOutputFormat.class);

        FileInputFormat.setInputPaths(job, new Path(strings[0]));
        FileOutputFormat.setOutputPath(job, new Path(strings[1]));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        private final static IntWritable ONE = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] strings = value.toString().split("\"");
            if (strings[1].contains("search?q")) {
                String[] b = strings[1].split("q=| ");
                String temp = b[2];
                //中文的话 做url decode
                temp = URLDecoder.decode(temp,"UTF-8");
                context.write(new Text(temp), ONE);
            }
        }
    }

    public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            Iterator<IntWritable> iterator = values.iterator();
            while (iterator.hasNext()) {
                sum = iterator.next().get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    private static void test() {
        String s = "192.168.1.100 - - [15/Sep/2015:00:11:23 +0800] \"GET /search?q=ssss HTTP/1.1\" 200 12";
        String[] strings = s.split("\"");
        System.out.println(Arrays.toString(strings));
        if (strings[1].contains("search?q")) {
            String[] b = strings[1].split("q=| ");
            System.out.println(Arrays.toString(b));
            System.out.println(b[2]);
        }
    }

}
